package wsf.akkascala

import akka.actor.{Actor, ActorRef}
import akkascala.{MapData, ReduceData, Word}

import scala.collection.mutable.HashMap

/**
 * Created by sjzx on 2016/5/13.
  * Reduce Actor receive mapdata then deal to reduce data then send to aggregate actor
 */
class ReduceWorker(aggreateActor: ActorRef) extends Actor{
  override def receive={
    case message: MapData => aggreateActor ! doReduce(message)
    case _ => unhandled()
  }

  def doReduce(data: MapData) : ReduceData = {
    val reduce = new HashMap[String, Int]
    for(d <- data.dataList) {
      if (reduce.contains(d.word)){
        val count = reduce.getOrElse(d.word, 1)
        reduce.put(d.word, count + 1)
      }
      else
        reduce.put(d.word, 1)
    }
    ReduceData(reduce)
  }
}
